/**
 * @author: yyq
 * @date: 2025年3月13日 10:09:19
 * @description: 
 */
package com.sheepone.monitor.client.handler.connect;

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.sheepone.monitor.anno.ChannelHandlerOrder;
import com.sheepone.monitor.enums.SerializationAlgorithmsEnum;
import com.sheepone.monitor.enums.SheeponePackageTypeEnum;
import com.sheepone.monitor.handler.AbstractSheeponeContentDuplexHandler;
import com.sheepone.monitor.proto.HeartBeatCheckMessage;
import com.sheepone.monitor.proto.HeartBeatReportMessage;
import com.sheepone.util.time.TimeFormatter;
import com.sheepone.webdev.log.WebDevLogger;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * @author: yyq
 * @date: 2025年3月13日 10:09:19
 * @description:
 *               1、每隔 1m 定时发送心跳包给Server
 *               2、负责将HeartBeatReportMessage转化成字节数组
 *               3、解析处理来自服务端的HeartBeatCheckMessage
 *               4、读写空闲超过5分钟发送心跳包给Server
 */
@ChannelHandlerOrder(Integer.MIN_VALUE + 1)
public class ClientHeartbeatHandler
        extends AbstractSheeponeContentDuplexHandler<HeartBeatCheckMessage, HeartBeatReportMessage> {
    public ClientHeartbeatHandler() {
        super(SheeponePackageTypeEnum.HEART_BEAT_CHECK, SheeponePackageTypeEnum.HEART_BEAT_REPORT,
                SerializationAlgorithmsEnum.PROTOBUF);
    }

    private static WebDevLogger logger = WebDevLogger.getLogger(ClientHeartbeatHandler.class);

    private ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
            1, new CustomThreadFactory("heartbeat-pool"));
    private long lastTime = -1;

    private void reportHeartBeat(Channel ch) {
        long current = System.currentTimeMillis();
        ch.writeAndFlush(HeartBeatReportMessage.newBuilder().setTime(current).build());
        logger.info("report heartbeat for {}", TimeFormatter.formatTimeStamp(current));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // ! channel建立连接后 每隔 1m 定时发送心跳包给server
        scheduler.scheduleAtFixedRate(() -> {
            Channel ch = ctx.channel();
            if (ch.isActive()) {
                this.reportHeartBeat(ch);
            }
        }, 1, 1, TimeUnit.MINUTES);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            if (((IdleStateEvent) evt).state().equals(IdleState.ALL_IDLE)) {
                this.reportHeartBeat(ctx.channel());
            }
        }
    }

    @Override
    public HeartBeatCheckMessage fromBytes(byte[] bytes, SerializationAlgorithmsEnum serialization) throws Exception {
        return HeartBeatCheckMessage.parseFrom(bytes);
    }

    @Override
    public byte[] toBytes(HeartBeatReportMessage content, SerializationAlgorithmsEnum serialization) throws Exception {
        return content.toByteArray();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HeartBeatCheckMessage msg) throws Exception {
        if (msg.getTime() > this.lastTime) {
            this.lastTime = msg.getTime();
            logger.info("receive heartbeat for {}", TimeFormatter.formatTimeStamp(msg.getTime()));
            this.reportHeartBeat(ctx.channel());
        }
    }
}

class CustomThreadFactory implements ThreadFactory {
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    CustomThreadFactory(String poolName) {
        namePrefix = poolName + "-thread-";
    }

    public Thread newThread(Runnable r) {
        return new Thread(r, namePrefix + threadNumber.getAndIncrement());
    }
}
